package com.xdhl.fu.email.consumer;

import com.alibaba.fastjson.JSONObject;
import com.xdhl.fu.common.core.constant.event.EmailSendConstants;
import com.xdhl.fu.common.core.utils.StringUtils;
import com.xdhl.fu.common.redis.service.RedisService;
import com.xdhl.fu.email.api.domain.TextEmail;
import com.xdhl.fu.email.api.enums.EmailType;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.mail.MailSendException;
import org.springframework.mail.SimpleMailMessage;
import org.springframework.mail.javamail.JavaMailSender;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * 发送邮件事件消费者
 *
 * @author 朱力维
 * @time 2022/6/15 11:37
 */
@Component
public class EmailSendEventConsumer {

    private static final Logger log = LoggerFactory.getLogger(EmailSendEventConsumer.class);

    public static final int DEFAULT_QUOTA = 50;

    public Map<String, Integer> quotas;

    @Value("#{'${spring.application.name}'+'-'+'${spring.profiles.active}'}")
    private String consumerGroup;

    @Value("${fu.rocketmq.namesrv-addr}")
    private String namesrvAddr;

    @Value("${spring.mail.username}")
    private String sender;

    private DefaultMQPushConsumer consumer;

    @Resource
    private RedisService redisService;

    @Resource
    JavaMailSender javaMailSender;

    @PostConstruct
    public void init() throws MQClientException {
        consumer = new DefaultMQPushConsumer(consumerGroup);
        consumer.setNamesrvAddr(namesrvAddr);

        //设置consumer所订阅的Topic和Tag，*代表全部的Tag
        consumer.subscribe(EmailSendConstants.TOPIC, "*");

        /**
         * CONSUME_FROM_LAST_OFFSET 默认策略，从该队列最尾开始消费，跳过历史消息
         * CONSUME_FROM_FIRST_OFFSET 从队列最开始开始消费，即历史消息（还储存在broker的）全部消费一遍
         */
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);

        consumer.registerMessageListener((MessageListenerConcurrently) (list, context) -> doing(list));
        consumer.start();

        // 初始化限额
        quotas = new HashMap<>();
        quotas.put(EmailType.VERIFICATION_CODE.name(), 100);

        log.info("发送邮件事件消费者启动成功。");
    }

    @PreDestroy
    public void destory() {
        consumer.shutdown();
        log.info("发送邮件事件消费者已销毁。");
    }

    public ConsumeConcurrentlyStatus doing(List<MessageExt> msgs) {
        ConsumeConcurrentlyStatus consumeConcurrentlyStatus;
        try {
            for (MessageExt msg : msgs) {
                TextEmail textEmail = JSONObject.parseObject(new String(msg.getBody(), RemotingHelper.DEFAULT_CHARSET), TextEmail.class);
                // 文本邮件
                if (!exceed(textEmail.getTo(), quotas.getOrDefault(textEmail.getEmailType().name(), DEFAULT_QUOTA))) {
                    SimpleMailMessage simpleMailMessage = new SimpleMailMessage();
                    simpleMailMessage.setFrom(sender);
                    simpleMailMessage.setTo(textEmail.getTo());
                    simpleMailMessage.setSubject(String.format("【迅盾互联】%s", textEmail.getSubject()));
                    simpleMailMessage.setText(textEmail.getText());
                    sendSimpleMailMessage(simpleMailMessage);
                    redisService.incr(getKey(textEmail.getTo()), 1);
                }
            }
            consumeConcurrentlyStatus = ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        } catch (Exception e) {
            log.error(e.getMessage());
            consumeConcurrentlyStatus = ConsumeConcurrentlyStatus.RECONSUME_LATER;
            // nothing to do.
        }
        return consumeConcurrentlyStatus;
    }

    /**
     * 获取缓存key
     *
     * @param to 收件人
     * @return
     */
    private String getKey(String to) {
        return "count:email:send:" + to;
    }

    /**
     * 发送简单邮件
     *
     * @param mailMessage 邮件对象
     */
    private void sendSimpleMailMessage(SimpleMailMessage mailMessage) {
        try {
            javaMailSender.send(mailMessage);
        } catch (MailSendException e) {
            log.error(e.getMessage());
        }
    }

    /**
     * 判断是否超出邮件发送限额
     *
     * @param key   邮件接收者
     * @param quota 限额
     * @return 是否超出
     */
    private boolean exceed(String key, Integer quota) {
        String value = redisService.getStringValue(getKey(key));
        if (StringUtils.isNotBlank(value)) {
            Integer intValue = Integer.valueOf(value);
            return intValue > quota;
        } else {
            return false;
        }
    }

}
